/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package selector

import (
	"bytes"
	"context"
	"encoding/json"
	"sort"
	"strconv"
	"sync"
	"time"

	"github.com/golang/protobuf/proto"
	"hundsun.com/hsl/hschain/common"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/util/pb"
	dposcomm "hundsun.com/hsl/hschain/consensus/impl/dpos/common"
	"hundsun.com/hsl/hschain/module/common/utils"
	"hundsun.com/hsl/hschain/protos"
	pbcommon "hundsun.com/hsl/hschain/protos/common"
	"hundsun.com/hsl/hschain/protos/consensus/dpos"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	"hundsun.com/hsl/hschain/store/db"
)

const (
	logFieldTxIndex = "txIndex"
)

// VoteCounter 统计投票结果
type VoteCounter struct {
	log                  logging.Log
	module               *DposSelector
	maxExecBlock         *MaxExecBlock
	sleepMillis          int
	blocksFetchBatchSize uint64
	sync.RWMutex
}

// MaxExecBlock 记录最大已执行块相关信息
type MaxExecBlock struct {
	round  int64
	number uint64
	exists bool
	sync.RWMutex
}

// Store 存储MaxExecBlock信息
func (m *MaxExecBlock) Store(exists bool, round int64, blockNumber uint64) {
	m.Lock()
	defer m.Unlock()
	m.exists = true
	m.round = round
	m.number = blockNumber
}

// Exists 是否存在最大执行区块
func (m *MaxExecBlock) Exists() bool {
	m.RLock()
	defer m.RUnlock()
	return m.exists
}

// GetRound 获取轮次信息
func (m *MaxExecBlock) GetRound() int64 {
	m.RLock()
	defer m.RUnlock()
	return m.round
}

// GetNumber 获取高度信息
func (m *MaxExecBlock) GetNumber() uint64 {
	m.RLock()
	defer m.RUnlock()
	return m.number
}

// Init 初始化
func (vc *VoteCounter) Init() error {
	vc.sleepMillis = utils.GetIntConfigParameterValue(vc.module.config, configParamVoteCounterSleepMillis, 100)
	vc.log.Infof("%s: %dms", configParamVoteCounterSleepMillis, vc.sleepMillis)

	blocksFetchBatchSize := utils.GetIntConfigParameterValue(vc.module.config, configParamVoteCounterBlocksFetchBatchSize, 10)
	if blocksFetchBatchSize <= 0 {
		blocksFetchBatchSize = 10
	}
	vc.blocksFetchBatchSize = uint64(blocksFetchBatchSize)
	vc.log.Infof("%s: %dms", configParamVoteCounterBlocksFetchBatchSize, vc.blocksFetchBatchSize)
	vc.maxExecBlock = &MaxExecBlock{}
	maxExecutedBlock, err := vc.module.client.RequestMaxExecutedBlock()
	if err != nil {
		return err
	}
	if maxExecutedBlock == nil {
		vc.maxExecBlock.Store(false, 0, 0)
	} else {
		roundOfMaxExecutedBlock, _, _ := dposcomm.CalculateRound(maxExecutedBlock.Block.Header.BlockTime)
		vc.maxExecBlock.Store(true, roundOfMaxExecutedBlock, maxExecutedBlock.BlockExecResult.BlockNumber)
	}
	//err = vc.subscribeBlockExecutedMessage()
	//if err != nil {
	//	return err
	//}
	return nil
}

//func (vc *VoteCounter) subscribeBlockExecutedMessage() error {
//	mqLog := vc.log.WithField(logging.LogFieldMessageTopic, message.TopicExecutedBlock)
//	mqLog.Infof("subscribing")
//	err := Client.Subscribe(message.TopicExecutedBlock, func(messageBytes []byte) {
//		if len(messageBytes) == 0 {
//			return
//		}
//
//		var blockExecMsg message.BlockExecutedMessage
//		err := json.Unmarshal(messageBytes, &blockExecMsg)
//		if err != nil {
//			mqLog.Warn("failed to unmarshal message")
//			return
//		}
//		blkLog := mqLog.WithField(logging.LogFieldMessageTopic, message.TopicExecutedBlock).
//			WithField(logging.LogFieldBlockNumber, blockExecMsg.BlockNumber)
//		roundOfBlock, _, _ := dposcomm.CalculateRound(blockExecMsg.BlockTime)
//		if !vc.maxExecBlock.Exists() || roundOfBlock > vc.maxExecBlock.GetRound() {
//			vc.maxExecBlock.Store(true, roundOfBlock, blockExecMsg.BlockNumber)
//			blkLog.Debugf("maxRoundOfExecutedBlocks set to '%d'", roundOfBlock)
//		} else {
//			blkLog.Debugf("round of message is the same as maxRoundOfExecutedBlocks: %d", roundOfBlock)
//		}
//	})
//	if err == nil {
//		mqLog.Info("subscribed")
//	}
//	return err
//}

// Run 开始统计选票
func (vc *VoteCounter) Run(ctx context.Context) {
	vc.log.Info("start dpos vote counter...")
	defer vc.log.Info("exit dpos vote counter...")

	lastHandleTime := time.Now().Unix()
	for {
		electionCtx := vc.module.electionProgress.Load()
		lastHandledBlock := electionCtx.LastHandledBlock
		roundInElectionCtx := electionCtx.Round
		select {
		case <-time.After(time.Duration(vc.sleepMillis) * time.Millisecond):
			if !vc.maxExecBlock.Exists() {
				vc.log.Debug("no block executed")
				continue
			}
			roundInBlockExecMsg := vc.maxExecBlock.GetRound()
			vc.log.Tracef("round  of BlockExecuteMessage: %d, round of ElectionContext: %d", roundInBlockExecMsg, roundInElectionCtx)
			if roundInBlockExecMsg > roundInElectionCtx {
				vc.log.Debugf("tryCountVotesOfBlocks,round  of BlockExecuteMessage: %d, round of ElectionContext: %d", roundInBlockExecMsg, roundInElectionCtx)
				err := vc.tryCountVotesOfBlocks(lastHandledBlock+1, vc.maxExecBlock.GetNumber()+1, roundInBlockExecMsg)
				if err != nil {
					vc.log.Error(err.Error())
				}
				lastHandleTime = time.Now().Unix()
			} else if time.Now().Unix()-lastHandleTime > int64(dposcomm.TimeSliceLength) {
				vc.log.Debugf("tryCountVotesOfBlocks,more than %d seconds passed since last count", dposcomm.TimeSliceLength)
				err := vc.tryCountVotesOfBlocks(lastHandledBlock+1, lastHandledBlock+uint64(dposcomm.MaxBlocksInTimeSlice)+1, roundInElectionCtx+1)
				if err != nil {
					vc.log.Error(err.Error())
				}
				lastHandleTime = time.Now().Unix()
			}
		case <-ctx.Done():
			return
		}

	}

}

// tryCountVotesOfBlocks 在上次统计结果基础之上尝试统计[startBlockNumber,endBlockNumber)范围内的投票，直到轮次大于或等于stopRound。
// 如果超过10块，则分批处理
func (vc *VoteCounter) tryCountVotesOfBlocks(startBlockNumber uint64, endBlockNumber uint64, stopRound int64) error {
	vc.log.Debugf("enter tryCountVotesOfBlocks,startBlockNumber:%d endBlockNumber:%d stopRound:%d", startBlockNumber, endBlockNumber, stopRound)
	defer vc.log.Debugf("exit tryCountVotesOfBlocks,startBlockNumber:%d endBlockNumber:%d stopRound:%d", startBlockNumber, endBlockNumber, stopRound)
	start := startBlockNumber
	var end uint64
	clonedElectionCtx := vc.module.electionProgress.Load()
	for {
		if endBlockNumber-start > vc.blocksFetchBatchSize {
			end = start + vc.blocksFetchBatchSize
		} else {
			end = endBlockNumber
		}
		vc.log.Debugf("fetching blocks[%d,%d)", start, end)
		unhandledBlocks, err := vc.module.client.RequestExecutedBlocks(start, end)
		if err != nil {
			return err
		}
		vc.log.Debugf("got %d blocks", len(unhandledBlocks))
		if len(unhandledBlocks) == 0 {
			return nil
		}

		for _, blk := range unhandledBlocks {
			blkRound, _, _ := dposcomm.CalculateRound(blk.Block.Header.BlockTime)
			blkLog := vc.log.WithField(logging.LogFieldBlockNumber, blk.Block.Header.Number).
				WithField(dposcomm.LogFieldRound, blkRound)

			if blkRound > clonedElectionCtx.Round {
				blkLog.Debugf("roundInElectionContext:%d startCounting", clonedElectionCtx.Round)
				clonedElectionCtx.Round = blkRound
				clonedElectionCtx.FirstBlockOfRound = blk.Block.Header.Number
				var minersOfNextRound = &dpos.RoundMiners{
					Round: blkRound + 1,
				}

				totalWeightOfCandidates := clonedElectionCtx.CountBallotsAndSweep(blkRound - 1)
				blkLog.Infof("candidates size of prev round is %d", len(totalWeightOfCandidates))
				if len(totalWeightOfCandidates) < dposcomm.MaxMinersCount {
					minersOfNextRound.Miners = vc.module.client.GetCurrentMiners()
					blkLog.Info("adopt genesis miners")
				} else {
					candidates := sortCandidates(totalWeightOfCandidates)
					candidates = candidates[:dposcomm.MaxMinersCount]
					minersOfNextRound.Miners = make([]*pbcommon.Miner, dposcomm.MaxMinersCount)
					for i := 0; i < dposcomm.MaxMinersCount; i++ {
						minersOfNextRound.Miners[i] = &pbcommon.Miner{
							Address: string(candidates[i].Address),
							PeerID:  clonedElectionCtx.Candidates[candidates[i].Address].PeerID,
						}
					}
					if blkLog.IsDebugEnabled() {
						for i, miner := range minersOfNextRound.Miners {
							blkLog.Debugf("miners[%d] of nextRound:%s ", i, miner.Address)
						}
					}

				}
				blkLog.Debug("start saving miners of next round to es")
				roundMinersKey := dposcomm.GenerateRoundMinersKey(minersOfNextRound.Round)
				nextRoundMinersInfo, err := proto.Marshal(minersOfNextRound)
				if err != nil {
					return err
				}
				vc.module.GetLedgerDB().Set(roundMinersKey, nextRoundMinersInfo)
				if err != nil {
					return err
				}
				blkLog.Debug("finish saving miners of next round to es")
			}
			err = vc.addBallotsOfBlockToElectionContext(blk, clonedElectionCtx)
			if err != nil {
				return err
			}

			if blkRound > vc.module.electionProgress.Load().Round {
				electionContextBytes, err := json.Marshal(clonedElectionCtx)
				if err != nil {
					vc.log.Errorf("Marshal election context info failed, err:%s\n", err.Error())
					return err
				}
				err = vc.module.client.StoreElectionContextInfo(electionContextBytes)
				if err != nil {
					return err
				}
				blkLog.Debug("published")
				if blkRound >= stopRound {
					blkLog.Debugf("return reason: round of block is greater than or equal to stopRound '%d'", stopRound)
					break
				}
			}
			clonedElectionCtx.LastHandledBlock = blk.Block.Header.Number
		}

		if end == endBlockNumber {
			vc.log.Debugf("return reason: reached endBlockNumber '%d'", end)
			break
		}
		start = end
	}
	vc.module.electionProgress.Store(clonedElectionCtx)
	return nil
}

func sortCandidates(totalWeightOfCandidates map[protos.Address]uint64) []*WeightedAddress {
	if len(totalWeightOfCandidates) == 0 {
		return nil
	}
	var candidates []*WeightedAddress
	for k, v := range totalWeightOfCandidates {
		candidates = append(candidates, &WeightedAddress{
			Address: k,
			Weight:  v,
		})
	}
	sort.Slice(candidates, func(i, j int) bool {
		if candidates[i].Weight == candidates[j].Weight {
			return candidates[i].Address < candidates[j].Address
		}
		return candidates[i].Weight > candidates[j].Weight
	})
	return candidates
}

// addBallotsOfBlockToElectionContext 将指定块中的选票添加到ElectionContext中
func (vc *VoteCounter) addBallotsOfBlockToElectionContext(blk *pbledger.ExecutedBlock, electionCtx *ElectionContext) error {
	ballotsOfVoters := vc.extractBallots(blk)
	if len(ballotsOfVoters) == 0 {
		return nil
	}
	blkNumber := blk.Block.Header.Number
	for _, ballot := range ballotsOfVoters {
		electionCtx.AddBallot(ballot, blkNumber)
	}
	return nil
}

// extractBallots 从区块中抽取投票交易，交易必须满足如下条件：
// 1）交易有效；
// 2) to是BlackHoleAddress
// 3) 交易执行结果中能找到namespace为BlackHoleAddress且key为SpecialWriteSetKeyFromAccountBalance的KV对，
//    并且Value必须能转换成uint64
// 4）payload不为空，且能转换成BlackHoleTransactionPayload消息,
//    BlackHoleTransactionPayload:type取值为1，payload能转成Ballots消息

func (vc *VoteCounter) extractBallots(blk *pbledger.ExecutedBlock) []*WeightedBallot {
	blkLog := pb.GetBlockLog(vc.log, blk.Block)
	blkLog.Debug("start extracting ballots")
	defer blkLog.Debug("finish extracting ballots")
	var result []*WeightedBallot
	for _, txExecResults := range blk.BlockExecResult.TransactionExecResults {
		txIndex := txExecResults.Receipt.TransactionIndex
		txLog := blkLog.WithField(logging.LogFieldTxHash, protos.Hash(txExecResults.Receipt.TxHash).GetShortString()).
			WithField(logFieldTxIndex, txIndex)
		if txExecResults.Receipt.TxStatus != pbledger.TransactionReceipt_SUCCESSFUL {
			txLog.Tracef("unsuccessful[txStatus=%d],ignored", txExecResults.Receipt.TxStatus)
			continue
		}

		tx := blk.Block.Transactions[txIndex]
		if tx.To != string(common.BlackHoleAddress) {
			txLog.Tracef("tx.To is not '%s'", common.BlackHoleAddress)
			continue
		}

		if len(tx.Payload) == 0 {
			txLog.Debug("payload is empty")
			continue
		}

		if txExecResults == nil {
			txLog.Warn("got BlackHole receiver,but no exec result found")
			continue
		}

		var fromAccountBalanceKV *pbledger.NsKV
		for _, kvw := range txExecResults.Kvs {
			if bytes.Equal(kvw.Key, db.GenDBKey([]byte(common.BlackHoleAddress), []byte(common.SpecialWriteSetKeyFromAccountBalance))) {
				fromAccountBalanceKV = &pbledger.NsKV{
					Namespace: string(common.BlackHoleAddress),
					Key:       common.SpecialWriteSetKeyFromAccountBalance,
					Value:     kvw.Value,
				}
				break
			}
		}
		if fromAccountBalanceKV == nil {
			txLog.Warn("no 'fromAccountBalance' KV found in vote transaction result")
			continue
		}
		weight, err := strconv.ParseUint(string(fromAccountBalanceKV.Value), 10, 64)
		if err != nil {
			txLog.Warn("failed to parse uint64 balance of 'from' of vote transaction")
			continue
		}

		var txPayload pbcommon.BlackHoleTransactionPayload
		if err := proto.Unmarshal(tx.Payload, &txPayload); err != nil {
			txLog.Warn("failed to parse payload of vote transaction to BlackHoleTransactionPayload")
			continue
		}
		if txPayload.Type != uint32(dpos.BlackHoleTransactionPayloadTypeBallots) {
			txLog.Warn("type of BlackHoleTransactionPayload in transaction is not BlackHoleTransactionPayloadTypeBallots")
			continue
		}
		var ballots dpos.Ballots
		if err := proto.Unmarshal(txPayload.Payload, &ballots); err != nil {
			txLog.Warn("payload of BlackHoleTransactionPayload in transaction is not Ballots")
			continue
		}

		for _, ballot := range ballots.Ballots {
			result = append(result, &WeightedBallot{
				VoterAddress: protos.Address(tx.From),
				Ballot:       ballot,
				Weight:       weight,
			})
			txLog.Debugf("found Ballot[voterAddress=%s,weight=%d,type=%d]", tx.From, weight, ballot.Type)
		}
	}
	return result
}

// WeightedAddress 带有权重的地址
type WeightedAddress struct {
	Address protos.Address
	Weight  uint64
}
